/*
 * TOP SECRET Copyright 2006-2015 Transsion.com All right reserved. This software is the confidential and proprietary
 * information of Transsion.com ("Confidential Information"). You shall not disclose such Confidential Information and
 * shall use it only in accordance with the terms of the license agreement you entered into with Transsion.com.
 */
package com.palmplaystore.bigdata.storm.appupdate;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Service;

import com.palmplaystore.bigdata.redis.RedisCluHelper;
import com.palmplaystore.bigdata.storm.appupdate.bean.AppObject;
import com.palmplaystore.bigdata.util.StringUtil;

/**
 * ClassName:AppHBaseService <br/>
 * Date: 2018年9月4日 下午4:48:06 <br/>
 * 
 * @author fenglibin
 * @version
 * @see
 */
@Service("appHBaseService")
public class AppHBaseService {

    private Logger logger = LoggerFactory.getLogger(AppHBaseService.class);

    @Value("#{settings['hbase.app_update.table_name']}")
    private String appUpdateTable;
    @Value("#{settings['hbase.app_install.table_name']}")
    private String appInstallTable;

    @Autowired
    HbaseTemplate  hbaseTemplate;
    @Autowired
    RedisCluHelper  redisCluHelper;

    @SuppressWarnings("unchecked")
    public void save(final AppObject appObject) {
        String htableName = null;
        if(appObject.getVersion()==-1) {//所有的App列表，不是单条的
            htableName = appUpdateTable;
        }else {
            htableName = appInstallTable;
        }
        final String htableNameFinal = htableName;
        final String appUpdateTableFinal = appUpdateTable;
        final String appInstallTableFinal = appInstallTable;
         
        hbaseTemplate.execute(htableName, new TableCallback<Object>() {
            public Object doInTable(HTableInterface htable) throws Throwable {
                String rowKey = appObject.getImei();
                Put put = null;
                if(htableNameFinal.equals(appInstallTableFinal)) {
                    rowKey = new StringBuilder(rowKey).append("_").append(appObject.getKeyIndex()).toString();
                    put = new Put(Bytes.toBytes(rowKey));
                    put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("imei"), Bytes.toBytes(appObject.getImei()));
                    if(!StringUtil.isEmpty(appObject.getPackageName())) {
                    	put.addColumn(Bytes.toBytes("app"), Bytes.toBytes("packagename"), Bytes.toBytes(appObject.getPackageName()));
                    }
                    if(!StringUtil.isEmpty(appObject.getVersion()+"")) {
                    	put.addColumn(Bytes.toBytes("app"), Bytes.toBytes("version"), Bytes.toBytes(appObject.getVersion()+""));
                    }
                }else if(htableNameFinal.equals(appUpdateTableFinal)) {
                    put = new Put(Bytes.toBytes(rowKey));
                    if(!StringUtil.isEmpty(appObject.getBrand())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("brand"), Bytes.toBytes(appObject.getBrand()));
                    }
                    if(!StringUtil.isEmpty(appObject.getChannel())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("channel"),Bytes.toBytes(appObject.getChannel()));
                    }
                    put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("createTime"), Bytes.toBytes(appObject.getCreateTime().toString()));
                    if(!StringUtil.isEmpty(appObject.getCurClientVersionCode())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("curclientversioncode"),Bytes.toBytes(appObject.getCurClientVersionCode()));
                    }
                    if(!StringUtil.isEmpty(appObject.getImsi())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("imsi"), Bytes.toBytes(appObject.getImsi()));
                    }
                    if(!StringUtil.isEmpty(appObject.getLan())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("lan"), Bytes.toBytes(appObject.getLan()));
                    }
                    if(!StringUtil.isEmpty(appObject.getNetType())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("nettype"),	Bytes.toBytes(appObject.getNetType()));
                    }
                    if(!StringUtil.isEmpty(appObject.getSystemVersionCode())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("systemversioncode"),Bytes.toBytes(appObject.getSystemVersionCode()));
                    }
                    if(!StringUtil.isEmpty(appObject.getUa())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("ua"), Bytes.toBytes(appObject.getUa()));
                    }
                    if(!StringUtil.isEmpty(appObject.getCountryCode())) {
                    	put.addColumn(Bytes.toBytes("location"), Bytes.toBytes("countrycode"),Bytes.toBytes(appObject.getCountryCode()));
                    }
                    if(!StringUtil.isEmpty(appObject.getIpAddress())) {
                    	put.addColumn(Bytes.toBytes("location"), Bytes.toBytes("ipaddress"),Bytes.toBytes(appObject.getIpAddress()));
                    }
                    if(!StringUtil.isEmpty(appObject.getOperator())) {
                    	put.addColumn(Bytes.toBytes("location"), Bytes.toBytes("operator"),Bytes.toBytes(appObject.getOperator()));
                    }
                    put.addColumn(Bytes.toBytes("valid"), Bytes.toBytes("md5"), Bytes.toBytes(appObject.getMd5()));
                    put.addColumn(Bytes.toBytes("app"), Bytes.toBytes("apps"), Bytes.toBytes(appObject.getPackageName()));
                }
                htable.put(put);
                return null;
            }
        });
        logger.info("appObject=" + appObject.toString());
    }
    /**
     * 批量操作
     * @param appObjects 拆分后的appInstallTable
     * @param allAppObject 没有拆分的 appUpdateTable
     */
    @SuppressWarnings("unchecked")
    public boolean saveBatch(final List<AppObject> appObjects,final AppObject allAppObject) {
    	Integer allTotal=0;
    	String apps="";
    	if(!StringUtil.isEmpty(allAppObject.getIsUpdate())&&allAppObject.getIsUpdate().equals("1")) {
    		Map data=getData(appUpdateTable,allAppObject.getImei());
    		if(data!=null) {
    			if(!StringUtil.isEmpty(data.get("apptotal")+"")) {
    				allTotal=Integer.parseInt(data.get("apptotal")+"");
    			}
    			apps=data.get("apps")+"";
    		}
    	}
    	boolean installflag=processAppInstallTable(appObjects,allAppObject,allTotal);
    	if(!installflag) {
    		return installflag;
    	}
    	boolean updateflag=processAppUpdateTable(allAppObject,appObjects.size());
    	logger.warn("updateflag=" + updateflag+",installflag="+installflag);
    	return updateflag;
    }
    
    private Map getData(String tableName,String rowKey){
        Map t= hbaseTemplate.get(tableName, rowKey,new RowMapper<Map>() {
			@Override
			public Map mapRow(Result result, int rowNum) throws Exception {
				Map rs=new HashMap();
				rs.put("apps", Bytes.toString(result.getValue(Bytes.toBytes("app"), Bytes.toBytes("apps"))));
				rs.put("apptotal", Bytes.toString(result.getValue(Bytes.toBytes("app"), Bytes.toBytes("apptotal"))));
				return rs;
			}
            });
        return t;
    }
    private Boolean processAppInstallTable(final List<AppObject> appObjects,final AppObject allAppObject,final Integer allTotal){
    	boolean installflag=hbaseTemplate.execute(appInstallTable, new TableCallback<Boolean>() {
    		public Boolean doInTable(HTableInterface appInstallTable) throws Throwable {
    			try {
    				List<Put> list=new ArrayList<Put>();
    				String rk = allAppObject.getImei();
					try {
						rk=(Long.parseLong(rk)%10)+"_"+rk;
    				}catch(Exception e) {
    				}
    				if(!StringUtil.isEmpty(allAppObject.getIsUpdate())&&allAppObject.getIsUpdate().equals("1")&&allTotal>0) {//更新
    					List<Delete> deletes=new ArrayList<Delete>();
    					for(int i=1;i<=allTotal;i++) {
    						Delete delete = new Delete(Bytes.toBytes(rk+"_"+i));
    						deletes.add(delete);
    					}
    					appInstallTable.delete(deletes);
    					logger.warn("删除成功,size="+allTotal);
    				}
    				for(AppObject appObject:appObjects) {
    					String rowKey = new StringBuilder(rk).append("_").append(appObject.getKeyIndex()).toString();
    					Put  put = new Put(Bytes.toBytes(rowKey));
    					put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("imei"), Bytes.toBytes(appObject.getImei()));
    					put.addColumn(Bytes.toBytes("app"), Bytes.toBytes("packagename"), Bytes.toBytes(appObject.getPackageName()));
    					put.addColumn(Bytes.toBytes("app"), Bytes.toBytes("version"), Bytes.toBytes(appObject.getVersion()+""));
    					list.add(put);
    				}
    				appInstallTable.put(list);
    				return true;
    			}catch(Exception e) {
    				logger.error("写入"+appInstallTable+"异常"+e.getMessage(),e);
    				return false;
    			}
    		}
    	});
    	return installflag;
    }
    private Boolean processAppUpdateTable(final AppObject allAppObject,final Integer newTotal){
    	Boolean updateflag=hbaseTemplate.execute(appUpdateTable, new TableCallback<Boolean>() {
    		public Boolean doInTable(HTableInterface appUpdateTable) throws Throwable {
    			try {
    				AppObject appObject=allAppObject;
    				String rowKey = appObject.getImei();
    				try {
    					rowKey=(Long.parseLong(rowKey)%10)+"_"+rowKey;
    				}catch(Exception e) {
    				}
    				Put put = new Put(Bytes.toBytes(rowKey));
    				if(!StringUtil.isEmpty(appObject.getBrand())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("brand"), Bytes.toBytes(appObject.getBrand()));
                    }
                    if(!StringUtil.isEmpty(appObject.getChannel())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("channel"),Bytes.toBytes(appObject.getChannel()));
                    }
                    put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("createTime"), Bytes.toBytes(appObject.getCreateTime().toString()));
                    if(!StringUtil.isEmpty(appObject.getCurClientVersionCode())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("curclientversioncode"),Bytes.toBytes(appObject.getCurClientVersionCode()));
                    }
                    if(!StringUtil.isEmpty(appObject.getImsi())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("imsi"), Bytes.toBytes(appObject.getImsi()));
                    }
                    if(!StringUtil.isEmpty(appObject.getLan())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("lan"), Bytes.toBytes(appObject.getLan()));
                    }
                    if(!StringUtil.isEmpty(appObject.getNetType())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("nettype"),	Bytes.toBytes(appObject.getNetType()));
                    }
                    if(!StringUtil.isEmpty(appObject.getSystemVersionCode())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("systemversioncode"),Bytes.toBytes(appObject.getSystemVersionCode()));
                    }
                    if(!StringUtil.isEmpty(appObject.getUa())) {
                    	put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("ua"), Bytes.toBytes(appObject.getUa()));
                    }
                    if(!StringUtil.isEmpty(appObject.getCountryCode())) {
                    	put.addColumn(Bytes.toBytes("location"), Bytes.toBytes("countrycode"),Bytes.toBytes(appObject.getCountryCode()));
                    }
                    if(!StringUtil.isEmpty(appObject.getIpAddress())) {
                    	put.addColumn(Bytes.toBytes("location"), Bytes.toBytes("ipaddress"),Bytes.toBytes(appObject.getIpAddress()));
                    }
                    if(!StringUtil.isEmpty(appObject.getOperator())) {
                    	put.addColumn(Bytes.toBytes("location"), Bytes.toBytes("operator"),Bytes.toBytes(appObject.getOperator()));
                    }
    				put.addColumn(Bytes.toBytes("valid"), Bytes.toBytes("md5"), Bytes.toBytes(appObject.getMd5()));
    				put.addColumn(Bytes.toBytes("app"), Bytes.toBytes("apps"), Bytes.toBytes(appObject.getPackageName()));
    				put.addColumn(Bytes.toBytes("app"), Bytes.toBytes("apptotal"), Bytes.toBytes(newTotal+""));
    				appUpdateTable.put(put);
    				try {
    					redisCluHelper.del(rowKey);
    				}catch(Exception e) {
    					logger.error("写入redis失败"+e.getMessage(),e);
    				}
    				return true;
    			}catch(Exception e) {
    				logger.error("写入"+appUpdateTable+"异常"+e.getMessage(),e);
    				return false;
    			}
    		}
    	});
    	return updateflag;
    }
}
